package io.xxx.xbutler.core.event.job;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import io.xxx.xbutler.constant.Topics;
import io.xxx.xbutler.data.TaskMapper;
import io.xxx.xbutler.domain.task.Task;
import io.xxx.xbutler.domain.task.TaskStatus;
import jakarta.annotation.Nonnull;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.JobExecutionContext;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;

/**
 * @see io.xxx.xbutler.core.event.listen.TaskExecuteListener
 */
@Slf4j
@Component
@DisallowConcurrentExecution
public class TaskExecuteJob extends QuartzJobBean {

    @Resource
    private PulsarTemplate<Long> pulsarTemplate;

    @Resource
    private TaskMapper taskMapper;

    @Override
    protected void executeInternal(@Nonnull JobExecutionContext context) {
        LocalDateTime now = LocalDateTime.now();
        LambdaQueryWrapper<Task> wrapper = new LambdaQueryWrapper<Task>()
                .eq(Task::getStatus, TaskStatus.RUNNABLE)
                .le(Task::getStartTime, now)
                .gt(Task::getEndTime, now);
        List<Task> tasks = taskMapper.selectList(wrapper);

        for (Task task : tasks) {
            pulsarTemplate.send(Topics.TASK_EXECUTE, task.getId());
            task.setStatus(TaskStatus.FIRED);
            taskMapper.updateById(task);
            log.info("任务触发执行[{}]", task.getId());
        }
    }
}
